package com.dzx;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.*;
import java.net.URI;

/**
 * Hello world!
 */
public class HDFSApp {

    FileSystem fileSystem = null;

    Configuration configuration = null;

    public static final String HDFS_PATH = "hdfs://192.168.42.85:8020";

    @Test
    public void mkdir() throws Exception {
        fileSystem.mkdirs(new Path("/user/root"));
    }

    @Test
    public void create() throws Exception {
        FSDataOutputStream outputStream = fileSystem.create(new Path("/user/root/hello.txt"));
        outputStream.write("hello hadoop hello hadoop ok my state is ok you are a dog and pig".getBytes());
        outputStream.flush();
        outputStream.close();
    }

    /**
     * 查看文本文件
     *
     * @throws Exception
     */
    @Test
    public void cat() throws Exception {
        FSDataInputStream inputStream = fileSystem.open(new Path("/hdfsapi/test/a.txt"));
        IOUtils.copyBytes(inputStream, System.out, 1024);
        inputStream.close();
    }


    /**
     * copy form 本地
     *
     * @throws Exception
     */
    @Test
    public void copyFromLocalFile() throws Exception {
        Path localPath = new Path("D:/1.JPG");
        Path hdfsPath = new Path("/hdfsapi/test/1.JPG");
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }



    /**
     * copy form 本地
     *
     * @throws Exception
     */
    public void copyFromLocalFile(String localFile,String hdfsFile) throws Exception {
        Path localPath = new Path(localFile);
        Path hdfsPath = new Path(hdfsFile);
        fileSystem.copyFromLocalFile(localPath, hdfsPath);
    }


    /**
     * 上传文件带有进度
     *
     * @throws IOException
     */
    @Test
    public void uploadFileWHithProgress() throws IOException {

        InputStream in = new BufferedInputStream(new FileInputStream(new File("D:/jdk-8u171-linux-x64.tar.gz")));

        FSDataOutputStream fsDataOutputStream = fileSystem.create(new Path("/hdfsapi/test/jdk-8u171-linux-x64.tar.gz"), new Progressable() {
            @Override
            public void progress() {
                System.out.print(".");
            }
        });
        IOUtils.copyBytes(in, fsDataOutputStream, 4096);
    }

    /**
     * 从 hdfs 拷贝文件到本地
     *
     * @throws IOException
     */
    @Test
    public void copyToLocal() throws IOException {
        fileSystem.copyToLocalFile(new Path("/hdfsapi/test/1.JPG"), new Path("F:/2.JPG"));
    }

    /**
     * 重命名
     *
     * @throws Exception
     */
    @Test
    public void rename() throws Exception {
        Path oldPath = new Path("/hdfsapi/test/a.txt");
        Path newPath = new Path("/hdfsapi/test/b.txt");
        fileSystem.rename(oldPath, newPath);
    }


    @Test
    public void listFiles() throws Exception {
        FileStatus[] fileStatuses = fileSystem.listStatus(new Path("/"));
        for (FileStatus fileStatus : fileStatuses) {
            String dir = fileStatus.isDirectory() ? "目录" : "文件";
            long len = fileStatus.getLen();
            short replication = fileStatus.getReplication();
            String symlink = null;
            try {
                symlink = fileStatus.getSymlink().toString();
            } catch (Exception e) {

            }
            long blockSize = fileStatus.getBlockSize();
            long accessTime = fileStatus.getAccessTime();
            String group = fileStatus.getGroup();
            long modificationTime = fileStatus.getModificationTime();
            String owner = fileStatus.getOwner();
            String s = fileStatus.getPath().toString();
            System.out.println(dir + "\t" + len + "\t" + replication + "\t" + blockSize + "\t" +
                    accessTime + "\t" + group + "\t" + modificationTime + "\t" + owner + "\t" + s + "\t" + symlink);
        }
    }


    @Before
    public void setUp() throws Exception {
        System.out.println("hdfs.setup");
        configuration = new Configuration();
        fileSystem = FileSystem.get(new URI(HDFS_PATH), configuration, "root");
    }


    @After
    public void tearDown() {
        configuration = null;
        fileSystem = null;
        System.out.println("hdfs teardown");
    }
}

